package org.enhance.local.tq.factory;

import org.apache.commons.collections4.CollectionUtils;
import org.enhance.common.util.Assertion;
import org.enhance.local.tq.message.TaskObj;
import org.enhance.local.tq.task.ITqHandler;
import lombok.extern.slf4j.Slf4j;
import org.enhance.local.tq.task.TqThread;
import org.springframework.util.Assert;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

@Slf4j
public class TaskQueueClientFactory<T extends ITqHandler> {
    private TaskQueueClientFactory() {
    }

    public static TaskQueueClientFactory getInstance() {
        return TaskQueueClientFactoryInstance.instance;
    }

    private static class TaskQueueClientFactoryInstance {
        private static TaskQueueClientFactory instance = new TaskQueueClientFactory();
    }

    private List<T> handleList = new ArrayList<>();
    private static Map<String, LinkedBlockingQueue<TaskObj>> taskObjQueueMap = new HashMap<>();

    public static LinkedBlockingQueue<TaskObj> getTaskObjQueue(String taskType) {
        return taskObjQueueMap.get(taskType);
    }

    public void setHandler(List<T> handleList) {
        this.handleList = handleList;
        if(CollectionUtils.isNotEmpty(handleList)) {
            for (T handler : handleList) {
                taskObjQueueMap.put(handler.getTaskType(), new LinkedBlockingQueue<>());
            }
        }
    }
    // ---------------------- init destroy  ----------------------

    public void init() {
        // start task pull
        startTaskConsumer();
    }

    public void destroy() {
        // pre : destory ClientFactoryThreadPool
        destoryClientFactoryThreadPool();
    }

    // ---------------------- thread pool ----------------------

    private ExecutorService clientFactoryThreadPool = Executors.newCachedThreadPool();
    public static volatile boolean clientFactoryPoolStoped = false;

    /**
     * destory consumer thread
     */
    private void destoryClientFactoryThreadPool(){
        clientFactoryPoolStoped = true;
        clientFactoryThreadPool.shutdownNow();
    }

    // ---------------------- handler service ----------------------
    public static <S extends TaskObj> void addTask(S taskObj){
        LinkedBlockingQueue<TaskObj> queue = getTaskObjQueue(taskObj.getTaskType());
        Assertion.notNull(queue, "unsupport task type:" + taskObj.getTaskType());
        // async queue, mult send
        queue.add(taskObj);
    }

    private void startTaskConsumer() {
        if(CollectionUtils.isEmpty(handleList)) {
            return;
        }
        for (T handler : handleList) {
            for (int i = 0; i < 3; i++) {
                clientFactoryThreadPool.execute(new TqThread<>(handler));
            }
        }
    }
}
